package com.zss.rocketmq.producer.service.common.impl;

import com.alibaba.fastjson.JSON;
import com.zss.rocketmq.producer.common.responseEntity.ResultVO;
import com.zss.rocketmq.producer.service.common.RocketmqProducerService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class RocketmqProducerServiceImpl implements RocketmqProducerService {
    public static final Logger logger = LoggerFactory.getLogger(RocketmqProducerServiceImpl.class);
    @Autowired
    DefaultMQProducer defaultMQProducer;
    @Override
    public ResultVO<String> send(String topic, String tag, String msg) {
        logger.info("RocketmqProducerServiceImpl.send|topice:{},tag:{},msg:{}",topic,tag,msg);
        if (StringUtils.isBlank(msg)){
            logger.info("消息为空");
            return ResultVO.createSuccess();
        }
        Message sendMsg = new Message(topic, tag, msg.getBytes());
        SendResult sendResult = null;
        try {
            sendResult = defaultMQProducer.send(sendMsg);
            logger.info("rocketmq消息发送响应：" + sendResult.toString());
        } catch (MQClientException e) {
            logger.info("rocketmq客户端异常",e);
        } catch (RemotingException e) {
            logger.info("远程调用rocketmq异常",e);
        } catch (MQBrokerException e) {
            logger.info("远程调用rocketmq.broker异常",e);
        } catch (InterruptedException e) {
            logger.info("rocketmq工作线程被打断异常",e);
        }
        return ResultVO.createSuccess(JSON.toJSONString(sendResult));
    }

    @Override
    public ResultVO sendByHashKey(String topic, String tag, String hashKey, String msg) {
        logger.info("RocketmqProducerServiceImpl.sendByHashKey|topice:{},tag:{},msg:{},hashKey:{}",topic,tag,msg,hashKey);
        if (StringUtils.isBlank(msg)){
            logger.info("消息为空");
            return ResultVO.createSuccess();
        }
        Message sendMsg = new Message(topic, tag,hashKey, msg.getBytes());
        SendResult sendResult = null;
        try {
            sendResult = defaultMQProducer.send(sendMsg);
            logger.info("rocketmq消息发送响应：" + sendResult.toString());
        } catch (MQClientException e) {
            logger.info("rocketmq客户端异常",e);
        } catch (RemotingException e) {
            logger.info("远程调用rocketmq异常",e);
        } catch (MQBrokerException e) {
            logger.info("远程调用rocketmq.broker异常",e);
        } catch (InterruptedException e) {
            logger.info("rocketmq工作线程被打断异常",e);
        }
        return ResultVO.createSuccess(JSON.toJSONString(sendResult));
    }

    @Override
    public ResultVO sendByOrderly(String topic, String tag, String hashKey, String msg) {
        logger.info("RocketmqProducerServiceImpl.sendByOrderly|topice:{},tag:{},msg:{},hashKey:{}",topic,tag,msg,hashKey);
        if (StringUtils.isBlank(msg)){
            logger.info("消息为空");
            return ResultVO.createSuccess();
        }
        Message sendMsg = new Message(topic, tag,hashKey, msg.getBytes());
        SendResult sendResult = null;
        try {
            sendResult = defaultMQProducer.send(sendMsg,new MessageQueueSelector(){

                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {
                    Integer id = (Integer) arg;
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            },tag.hashCode());
            logger.info("rocketmq消息发送响应：" + sendResult.toString());
        } catch (MQClientException e) {
            logger.info("rocketmq客户端异常",e);
        } catch (RemotingException e) {
            logger.info("远程调用rocketmq异常",e);
        } catch (MQBrokerException e) {
            logger.info("远程调用rocketmq.broker异常",e);
        } catch (InterruptedException e) {
            logger.info("rocketmq工作线程被打断异常",e);
        }
        return ResultVO.createSuccess(JSON.toJSONString(sendResult));
    }
}
